//! High-level async interface to the low-level Sled message queue.

use std::sync::Arc;

use tokio::task;
use tracing::*;

use aspect_db_llmq::LlmqDatabase;
pub use aspect_db_mq_types::{QueueId, ReaderId};
use aspect_msgqueue::{Msg, QueueView};

use crate::errors::*;

pub struct AsyncMqDatastore {
    db: Arc<LlmqDatabase>,
}

impl AsyncMqDatastore {
    pub fn new(db: LlmqDatabase) -> Result<Self, Error> {
        Ok(Self { db: Arc::new(db) })
    }

    pub async fn create_mq(&self, readers: Vec<ReaderId>) -> Result<QueueId, Error> {
        let db = self.db.clone();
        let h = task::spawn_blocking(move || {
            let id = db.create_queue(readers)?;
            debug!(%id, "created new queue");
            Ok(id)
        });

        Ok(wait_on_handle(h, "create_mq").await?)
    }

    pub async fn subscribe_reader_to_queue(
        &self,
        queue_id: &QueueId,
        reader_id: &ReaderId,
    ) -> Result<bool, Error> {
        let db = self.db.clone();
        let qid = *queue_id;
        let rid = *reader_id;
        let h = task::spawn_blocking(move || Ok(db.subscribe_reader_to_queue(&qid, &rid)?));
        Ok(wait_on_handle(h, "subscribe_reader_to_queue").await?)
    }

    pub async fn subscribe_new_reader_to_queue(
        &self,
        queue_id: &QueueId,
    ) -> Result<ReaderId, Error> {
        let db = self.db.clone();
        let qid = *queue_id;
        let h = task::spawn_blocking(move || Ok(db.subscribe_new_reader_to_queue(&qid)?));
        Ok(wait_on_handle(h, "subscribe_new_reader_to_queue").await?)
    }

    pub async fn unsubscribe_reader_from_queue(
        &self,
        queue_id: &QueueId,
        reader_id: &ReaderId,
    ) -> Result<bool, Error> {
        let db = self.db.clone();
        let qid = *queue_id;
        let rid = *reader_id;
        let h = task::spawn_blocking(move || Ok(db.unsubscribe_reader_from_queue(&qid, &rid)?));
        Ok(wait_on_handle(h, "unsubscribe_reader_from_queue").await?)
    }

    pub async fn insert_msg_into_queue(&self, queue_id: &QueueId, msg: Msg) -> Result<u64, Error> {
        let db = self.db.clone();
        let qid = *queue_id;
        let h = task::spawn_blocking(move || {
            let idx = db.insert_msg_into_queue(&qid, &msg)?;
            // TODO log?
            Ok(idx)
        });

        Ok(wait_on_handle(h, "insert_msg_raw").await?)
    }

    pub async fn submit_generic_msg(
        &self,
        queue_id: &QueueId,
        body: Vec<u8>,
    ) -> Result<u64, Error> {
        let db = self.db.clone();
        let qid = *queue_id;
        let h = task::spawn_blocking(move || {
            let idx = db.submit_generic_msg(&qid, body)?;
            // TODO log?
            Ok(idx)
        });

        Ok(wait_on_handle(h, "insert_msg_generic").await?)
    }

    pub async fn query_queue_view(
        &self,
        queue_id: &QueueId,
        reader_id: Option<ReaderId>,
    ) -> Result<QueueView, Error> {
        let db = self.db.clone();
        let qid = *queue_id;
        let h = task::spawn_blocking(move || {
            let qv = db.query_queue_view(&qid, reader_id.as_ref())?;
            // TODO log?
            Ok(qv)
        });

        Ok(wait_on_handle(h, "query_queue_view").await?)
    }

    pub async fn query_queue_message_range(
        &self,
        queue_id: &QueueId,
        reader_id: &ReaderId,
        min: u64,
        max: u64,
    ) -> Result<Vec<(u64, Msg)>, Error> {
        let db = self.db.clone();
        let qid = *queue_id;
        let rid = *reader_id;
        let h = task::spawn_blocking(move || {
            let res = db.query_queue_message_range(&qid, &rid, min, max)?;
            // TODO log?
            Ok(res)
        });

        Ok(wait_on_handle(h, "query_queue_message_range").await?)
    }

    pub async fn update_consumed_position(
        &self,
        queue_id: &QueueId,
        reader_id: &ReaderId,
        target: u64,
    ) -> Result<(u64, u64), Error> {
        let db = self.db.clone();
        let qid = *queue_id;
        let rid = *reader_id;
        let h = task::spawn_blocking(move || {
            let res = db.update_consumed_position(&qid, &rid, target)?;
            // TODO log?
            Ok(res)
        });

        Ok(wait_on_handle(h, "update_consumed_position").await?)
    }
}

async fn wait_on_handle<T>(
    handle: task::JoinHandle<Result<T, Error>>,
    name: &'static str,
) -> Result<T, Error> {
    match handle.await {
        Ok(v) => Ok(v?),
        Err(e) => {
            if e.is_panic() {
                Err(Error::BlockingTaskPanic(name))
            } else {
                Err(Error::BlockingTaskAborted(name))
            }
        }
    }
}
